package com.nepu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.nepu.gmall.realtime.util.DateFormatUtil;
import com.nepu.gmall.realtime.util.KafkaUtils;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 本类做的是流量域的独立访客的计算，具体的计算的步骤是如下：
 *  （1）当前计算的是每日的独立访客，对于一访客的标识使用的是mid。
 *  （2）一个mid在每天可能会访问很多次，我们只需要保留其中的一条就可以了。所以这里需要对数据进行过滤，过滤的标准是，
 *      当前用户访问的页面的last_page_id 为null。
 *      但是这样的数据，是可能有很多的，我们只需要保留每日的第一条，所以我们在这里需要维护一个状态。
 *      状态存储的是，当日第一个mid的访问的时间。
 *      这样我们在保存状态的时候，就需要对状态进行判断，如果状态为null,直接将状态保存，如果状态不为null，就判断当前的时间
 *      和状态的时间是否相同，如果不同就更新状态，如果相同就将这条数据丢掉。
 *      这样做其实就可以满足需求，如果我们想要做的更好我们还可以加上状态的生存时间，这样可以对跨天对状态进行更新。
 *  （3）将数据写出
 *
 *  测试的时候数据的流向：
 *  web/app  -> ngnix -> 日志服务器 -> kafka -> BaseLogApp -> kafka -> DwdTrafficUniqueVisitorDetail -> kafka
 *  mock --> f1.sh --> kafka(zookeeper) --> BaseLogApp --> DwdTrafficUniqueVisitorDetail
 * @author chenshuaijun
 * @create 2023-02-24 15:42
 */
public class DwdTrafficUniqueVisitorDetail {

    public static void main(String[] args) throws Exception {

        // TODO 1、加载环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 设置checkpoint的信息：设置checkpoint的间隔是5分钟,并且checkpoint的级别是精确一次性
        /*env.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE);
        // 设置checkpoint的超时时间是10分钟
        env.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);
        // 设置外部检查点。可以将检查点的元数据信息定期写入外部系统，这样当job失败时，检查点不会被清除。这样如果job失败，可以从检查点恢复job。
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 设置checkpoint的重启的策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(10, Time.of(1L, TimeUnit.DAYS), Time.of(3L, TimeUnit.MINUTES)));
        // 设置两个checkpoint之间的最小的间隔时间
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
        // 设置状态后端: 设置状态后端为内存级别
        env.setStateBackend(new HashMapStateBackend());
        // 设置checkpoint的存储的路径
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/checkpoint");
        // 因为我们的HDFS只有atguigu用户才能够操作，所以要将用户设置为atguigu
        System.setProperty("HADOOP_USER_NAME", "atguigu");*/
        // TODO 2、从kafka中消费数据
        String topic = "dwd_traffic_page_log";
        String groupId = "DwdTrafficUniqueVisitorDetail";
        DataStreamSource<String> kafkaSourceStream = env.addSource(KafkaUtils.getKafkaConsumer(topic, groupId));
        // TODO 3、转换数据结构
        SingleOutputStreamOperator<JSONObject> jsonStream = kafkaSourceStream.map(JSON::parseObject);
        // TODO 4、对数据进行过滤
        SingleOutputStreamOperator<JSONObject> filterStream = jsonStream.filter(pageJson -> pageJson.getJSONObject("page").getString("last_page_id") == null);
        // TODO 5、对数据进行分组
        KeyedStream<JSONObject, String> keyedStream = filterStream.keyBy(filterData -> filterData.getJSONObject("common").getString("mid"));
        // TODO 6、维护状态，只写出一条数据
        SingleOutputStreamOperator<JSONObject> singleVisitStream = keyedStream.filter(new RichFilterFunction<JSONObject>() {
            private ValueState<String> lastVisitDate;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> lastVisitDateDescriptor = new ValueStateDescriptor<>("last_visit_date", String.class);
                // 如果我们不设置状态的过期时间，那么状态是一直有效的，这里虽然是对我们的业务逻辑没有影响，但是我们也可以对整个业务进行更好的优化，所以这里添加上TTL
                StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.days(1))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) // 这个设置我们的状态的过期的条件是，在创建或者写入的时候更改过期时间
                        .build();
                lastVisitDateDescriptor.enableTimeToLive(ttlConfig);
                lastVisitDate = getRuntimeContext().getState(lastVisitDateDescriptor);
            }

            @Override
            public boolean filter(JSONObject value) throws Exception {
                // 1、首先取出状态
                String lastState = lastVisitDate.value();
                // 2、取出数据中的时间
                Long currentTs = value.getLong("ts");
                // 3、转换时间为年月日
                String currentDate = DateFormatUtil.toDate(currentTs);
                // 4、判断状态是否为null，或者状态和当前的数据时间是否相同
                if (lastState == null || !lastState.equals(currentDate)) {
                    // 更新状态
                    lastVisitDate.update(currentDate);

                    return true;
                } else {
                    return false;
                }
            }
        });
        // TODO 7、将数据写入到kafka对应的主题中
        singleVisitStream.print("独立访客数据-->");
        String targetTopic = "dwd_traffic_unique_visitor_detail";
        singleVisitStream.map(JSONAware::toJSONString).addSink(KafkaUtils.getFlinkKafkaProducer(targetTopic));

        // TODO 8、执行程序
        env.execute();
    }

}
